package app.dws;

import app.BaseAppV1;
import beans.ExamPaperBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.AtguiguUtil;
import utils.FlinkSinkUtil;

import java.math.BigDecimal;
import java.time.Duration;

public class Dws_15_ExamPaperCountWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_15_ExamPaperCountWindow().init(
                5011,
                2,
                "Dws_151_ExamPaperCountWindow",
                Constant.DWD_EXAM_PAPER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 将数据转换成POJO
        SingleOutputStreamOperator<ExamPaperBean> beanStream = parseToObj(stream);

        //对数据分组 开窗 聚合
        SingleOutputStreamOperator<ExamPaperBean> windowAndAggedStream = windowAndAgg(beanStream);

        //将数据写入ClickHouse
        writeToClickhouse(windowAndAggedStream);
    }

    private void writeToClickhouse(SingleOutputStreamOperator<ExamPaperBean> windowAndAggedStream) {
        windowAndAggedStream.addSink(FlinkSinkUtil.getClickHouseSink(Constant.DWS_EXAM_PAPER_SCORE_WINDOW,ExamPaperBean.class));
    }

    private SingleOutputStreamOperator<ExamPaperBean> windowAndAgg(SingleOutputStreamOperator<ExamPaperBean> beanStream) {
       return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ExamPaperBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner( (examPaperBean, ts) -> examPaperBean.getTs()))
                .keyBy(examPaperBean -> examPaperBean.getPaperId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ExamPaperBean>() {
                            @Override
                            public ExamPaperBean reduce(ExamPaperBean value1, ExamPaperBean value2) throws Exception {
                                value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                                value1.setScoreCount(value1.getScoreCount().add(value2.getScoreCount()));
                                value1.setDurCount(value1.getDurCount() + value2.getDurCount());
                                return value1;
                            }
                        }, new ProcessWindowFunction<ExamPaperBean, ExamPaperBean, Long, TimeWindow>() {
                            @Override
                            public void process(Long aLong,
                                                Context context,
                                                Iterable<ExamPaperBean> elements,
                                                Collector<ExamPaperBean> out) throws Exception {
                                ExamPaperBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());

                                out.collect(bean);

                            }
                        }
                );

    }

    private SingleOutputStreamOperator<ExamPaperBean> parseToObj(DataStreamSource<String> stream) {
        return stream
                .map(new RichMapFunction<String, ExamPaperBean>() {
                    @Override
                    public ExamPaperBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        Long paperId = jsonObject.getLong("paper_id");
                        String courseId = jsonObject.getString("course_id");
                        String paperTitle = jsonObject.getString("paper_title");
                        Long userCount = 1L;
                        BigDecimal scoreCount = jsonObject.getBigDecimal("score");
                        Long durCount = jsonObject.getLong("duration_sec");
                        return new ExamPaperBean("","",paperId,courseId,paperTitle,userCount,scoreCount,durCount, AtguiguUtil.dateToTs(jsonObject.getString("submit_time")));
                    }
                });
    }
}













